/*
 * MemPageBoundedBufferFixedKeyFixedValueImpl_.h
 *
 *  Created on: Aug 1, 2011
 *      Author: yong
 */

#ifndef MEMPAGEBOUNDEDBUFFERFIXEDKEYFIXEDVALUEIMPL__H_
#define MEMPAGEBOUNDEDBUFFERFIXEDKEYFIXEDVALUEIMPL__H_

#include<vector>
#include<cstdlib>
#include<boost/utility.hpp>
#include<boost/thread/condition.hpp>
#include<boost/thread/thread.hpp>
#include"../Typedef.h"

namespace distrim
{
/**
 * Fixed key size, fixed value size.
 */
template<typename K, typename V>
class MemPageBoundedBuffer<K, V, true, true> : private boost::noncopyable
{
public:
	typedef size_t index_type;

public:
	/**
	 * @param  initBufferCount  Initial buffer number
	 * @throws  std::runtime_error  If memory allocation failed.
	 */
	MemPageBoundedBuffer(size_t initBufferCount, size_t maxPairCount,
			size_t valueElemCount = 1) throw (std::runtime_error) :
		m_vecCircularBuff(initBufferCount), m_valueElemCount(valueElemCount),
				m_maxPairCount(maxPairCount), m_allocableBuffer(0),
				m_filledBuffer(0), m_producerThread(0), m_pageBytes((sizeof(K*)
						+ sizeof(K) + sizeof(V) * m_valueElemCount)
						* m_maxPairCount)
	{
		for (size_t i = 0; i < m_vecCircularBuff.size(); ++i)
		{
			m_vecCircularBuff[i].PageBegin = malloc(m_pageBytes);
			if (m_vecCircularBuff[i].PageBegin == NULL)
			{
				for (size_t j = 0; j < i; ++j)
				{
					free(m_vecCircularBuff[i].PageBegin);
				}
				throw std::runtime_error("can not allocate space");
			}
		}
	}
	MemPageBoundedBuffer()
	{
	}
	/**
	 * Recycle buffers allocated.
	 */
	virtual ~MemPageBoundedBuffer()
	{
		for (size_t i = 0; i < m_vecCircularBuff.size(); ++i)
		{
			free(m_vecCircularBuff[i].PageBegin);
		}
	}

	/**
	 * @precondition
	 * @throws   runtime_error   if memory allocation failed.
	 */
	inline void ReallocateBuffer(size_t bufferCount) throw (std::runtime_error)
	{
		if (bufferCount > m_vecCircularBuff.size() && m_producerThread == 0
				&& m_filledBuffer == 0)
		{
			size_t countToAdd = m_vecCircularBuff.size() - bufferCount;
			size_t pageBytes = (sizeof(K*) + sizeof(K) + sizeof(V)
					* m_valueElemCount) * m_maxPairCount;
			for (size_t i = 0; i < countToAdd; ++i)
			{
				// Allocate space for vector elements.
				PageInfo pi;
				// Allocate space for pages.
				pi.PageBegin = malloc(pageBytes);
				if (pi.PageBegin == NULL)
				{
					for (size_t j = 0; j < m_vecCircularBuff.size(); ++j)
					{
						free(m_vecCircularBuff[i].PageBegin);
					}
					throw std::runtime_error("can not allocate space");
				}
				m_vecCircularBuff.push_back(pi);
			}
			m_allocableBuffer = m_vecCircularBuff.size();
		}
	}

	/**
	 *
	 */
	index_type GetIndexToFill()
	{
		++m_producerThread;
		// Wait until find.
		while (true)
		{
			Lock lk(m_allocatableBufferMutex);
			if (m_allocableBuffer == 0)
			{
				m_allocatableBufferCondition.wait(lk);
			}
			else
			{
				for (size_t i = 0; i < m_vecCircularBuff.size(); ++i)
				{
					if (m_vecCircularBuff[i].State == StorageState::Empty)
					{
						m_vecCircularBuff[i].State = StorageState::Allocated;
						--m_allocableBuffer;
						return i;
					}
				}
			}
		}
	}

	/**
	 *
	 */
	void FillBuffer(index_type indexToFill, KeyValuePage<K, V> *kvPage)
	{
		// Fill buffer allocated.
		kvPage->SetPageBegin(m_vecCircularBuff[indexToFill].PageBegin);
		kvPage->Produce();
		m_vecCircularBuff[indexToFill].State = StorageState::Filled;
		Lock lk(m_filledBufferMutex);
		if (++m_filledBuffer == 1)
		{
			m_filledBufferCondition.notify_one();
		}
		{
			Lock lk(m_producerThreadMutex);
			if (--m_producerThread > 0)
			{
				m_allocatableBufferCondition.notify_all();
			}
		}
		m_filledBufferCondition.notify_all();
	}

	/**
	 * @return: get index of an empty buffer.
	 * Note: if all buffer have been processed, return -1.
	 */
	index_type GetIndexToEmpty()
	{
		while (true)
		{
			Lock lk(m_filledBufferMutex);
			if (m_filledBuffer > 0)
			{
				for (size_t i = 0; i < m_vecCircularBuff.size(); ++i)
				{
					if (m_vecCircularBuff[i].State == StorageState::Filled)
					{
						--m_filledBuffer;
						return i;
					}
				}
			}
			else if (m_filledBuffer == 0 && m_producerThread > 0)
			{
				m_filledBufferCondition.wait(lk);
			}
			else
			{
				return -1;
			}
		}
	}

	void EmptyBuffer(index_type indexToEmpty, KeyValuePage<K, V> *kvPage)
	{
		kvPage->SetPageBegin(m_vecCircularBuff[indexToEmpty].PageBegin);
		kvPage->Consume();
		m_vecCircularBuff[indexToEmpty].State = StorageState::Empty;
		{
			Lock lk(m_allocatableBufferMutex);
			if (++m_allocableBuffer == 1)
			{
				m_allocatableBufferCondition.notify_one();
			}
		}
	}

protected:
	std::vector<PageInfo> m_vecCircularBuff;
private:
	size_t m_valueElemCount, m_maxPairCount;
	size_t m_allocableBuffer, m_filledBuffer, m_producerThread;
	Mutex m_filledBufferMutex, m_allocatableBufferMutex, m_producerThreadMutex;
	Condition m_allocatableBufferCondition, m_filledBufferCondition,
			m_producerThreadCondition;
	size_t m_pageBytes;

};
}


#endif /* MEMPAGEBOUNDEDBUFFERFIXEDKEYFIXEDVALUEIMPL__H_ */
